package com.abyss.window;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * 滑动计数窗口案例
 *  每隔5条统计数据, 统计前10行
 *   1. 全量数字之和
 *   2. 分组后每个key对应的数字之和
 *
 *   窗口长度5, 滑动距离10
 */
public class SlidingCountWindowDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Source
        DataStreamSource<Tuple2<String, Integer>> source = env.addSource(new TumblingCountWindowDemo.GenerateRandomNumEverySecond());

        // 统计全部数据, 每隔5条, 统计前10条数据
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumAll = source
                .countWindowAll(10, 5).sum(1);

        // 统计每个key分组后的数据, 每隔5条
        // 这个5条是每个分组内达到5条, 而不是总的达到5条
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumEachKey = source
                .keyBy(0).countWindow(10, 5).sum(1);

        sumAll.print("Sum all:");
        sumEachKey.print("Sum each key:");

        env.execute();
    }

    /*
    自定义Source
    每隔1秒产生一个的k,v  k是hadoop spark flink 其中某一个, v是随机数字
     */
    public static class GenerateRandomNumEverySecond implements SourceFunction<Tuple2<String, Integer>> {
        private boolean isRun = true;
        private final Random random = new Random();
        private final List<String> keyList = Arrays.asList("hadoop", "spark", "flink");
        @Override
        public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
            while (this.isRun) {
                String key = keyList.get(random.nextInt(3));
                Tuple2<String, Integer> value = Tuple2.of(key, random.nextInt(9));
                ctx.collect(value);
                System.out.println("------: " + value);
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            this.isRun = false;
        }
    }
}